//程序入口
#include "main.h"
static struct shm_param para_modbus;
static struct shm_param para_stm32;

struct modbus_data *modbus_addr;
struct stm32_data *stm32_addr;
struct all_data_cache data_cache = {25.00, 25.00, 50.00, 0, 0, 0, 0, 50.00};
struct msgbuf send_buf;

MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;

//从共享内存读取数据并生成json返回
char *get_all_data()
{
    //创建一个json对象
    cJSON *json = cJSON_CreateObject();
    cJSON_AddItemToObject(json, "type", cJSON_CreateNumber(1));   //采集回复指令
    cJSON_AddItemToObject(json, "result", cJSON_CreateNumber(0)); //成功返回0，失败返回1
    cJSON *array = NULL;
    cJSON_AddItemToObject(json, "data", array = cJSON_CreateArray());

    //向json存入modbus数据
    char val[32] = "";
    cJSON *data_json = cJSON_CreateObject();
    snprintf(val, 32, "%.2lf", modbus_addr->temperature);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(101));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    data_json = cJSON_CreateObject();
    snprintf(val, 32, "%.2lf", modbus_addr->humidity);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(102));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    data_json = cJSON_CreateObject();
    snprintf(val, 32, "%d", modbus_addr->air_switch);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(103));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    data_json = cJSON_CreateObject();
    snprintf(val, 32, "%.2lf", modbus_addr->air_temp);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(104));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    data_json = cJSON_CreateObject();
    snprintf(val, 32, "%d", modbus_addr->light);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(105));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    //向json存入stm32数据
    data_json = cJSON_CreateObject();
    snprintf(val, 32, "%d", stm32_addr->led);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(301));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    data_json = cJSON_CreateObject();
    snprintf(val, 32, "%d", stm32_addr->Light);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(303));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    data_json = cJSON_CreateObject();
    snprintf(val, 32, "%.2lf", stm32_addr->bat);
    cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(302));
    cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
    cJSON_AddItemToArray(array, data_json);

    return cJSON_Print(json);
}

//从共享内存读取变化的数据并生成json返回
char *get_change_data(int *flag)
{
    //创建一个json对象
    cJSON *json = cJSON_CreateObject();
    cJSON_AddItemToObject(json, "type", cJSON_CreateNumber(1));   //采集回复指令
    cJSON_AddItemToObject(json, "result", cJSON_CreateNumber(0)); //成功返回0，失败返回1
    cJSON *array = NULL;
    cJSON_AddItemToObject(json, "data", array = cJSON_CreateArray());

    //向json存入modbus数据
    char val[32] = "";
    cJSON *data_json = NULL;
    if (data_cache.temperature != modbus_addr->temperature)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%.2lf", modbus_addr->temperature);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(101));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.temperature = modbus_addr->temperature;
        *flag = 1;
    }
    if (data_cache.humidity != modbus_addr->humidity)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%.2lf", modbus_addr->humidity);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(102));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.humidity = modbus_addr->humidity;
        *flag = 1;
    }
    if (data_cache.air_switch != modbus_addr->air_switch)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%d", modbus_addr->air_switch);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(103));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.air_switch = modbus_addr->air_switch;
        *flag = 1;
    }
    if (data_cache.air_temp != modbus_addr->air_temp)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%.2lf", modbus_addr->air_temp);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(104));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.air_temp = modbus_addr->air_temp;
        *flag = 1;
    }
    if (data_cache.light != modbus_addr->light)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%d", modbus_addr->light);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(105));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.light = modbus_addr->light;
        *flag = 1;
    }

    if (data_cache.led != stm32_addr->led)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%d", stm32_addr->led);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(301));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.led = stm32_addr->led;
        *flag = 1;
    }
    if (data_cache.Light != stm32_addr->Light)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%d", stm32_addr->Light);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(303));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.Light = stm32_addr->Light;
        *flag = 1;
    }
    if (data_cache.bat != stm32_addr->bat)
    {
        data_json = cJSON_CreateObject();
        snprintf(val, 32, "%.2lf", stm32_addr->bat);
        cJSON_AddItemToObject(data_json, "key", cJSON_CreateNumber(302));
        cJSON_AddItemToObject(data_json, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(array, data_json);
        data_cache.bat = stm32_addr->bat;
        *flag = 1;
    }
    return cJSON_Print(json);
}

//数据上报线程：周期上报，变化上报
void *handler_up()
{
    char send_buf[2048] = {0};
    while (1)
    {
        //读取上报类型
        int fd;
        char buf[2048] = {0};
        fd = open("/mnt/config/node.json", O_RDONLY);
        if (fd < 0)
        {
            perror("error");
            return NULL;
        }
        int read_len = read(fd, buf, 2048);
        if (read_len < 0)
        {
            perror("error");
            return NULL;
        }

        close(fd);
        //创建一个json对象
        cJSON *root = NULL;
        cJSON *item = NULL; //cjson对象
        root = cJSON_Parse(buf);
        item = cJSON_GetObjectItem(root, "report");
        int up_type = cJSON_GetObjectItem(item, "type")->valueint; //获取上报类型
        int flag = 0;                                              //变化上报的标志位  0-数据没有变化，不需要发送  1-数据发生了变化，需要发送

        if (up_type == 2)
        {
            //printf("周期上报\n");
            int sleep_time = cJSON_GetObjectItem(item, "period")->valueint;
            snprintf(send_buf, 2048, "%s", get_all_data());
            //printf("%s\n", send_buf);

            pubmsg.payload = send_buf;
            pubmsg.payloadlen = (int)strlen(send_buf);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, TOPIC_DATA_UP, &pubmsg, &token);
            rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
            printf("Message with delivery token %d delivered\n", token);
            sleep(sleep_time); //可进行优化，单独开辟一个线程读取上报类型的值，这样就不需要等待延时结束再判断下次上报的类型了
        }
        else if (up_type == 1)
        {
            //printf("变化上报\n");
            snprintf(send_buf, 2048, "%s", get_change_data(&flag));
            //发布数据
            pubmsg.payload = send_buf;
            pubmsg.payloadlen = (int)strlen(send_buf);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            if (flag == 1)
            {
                MQTTClient_publishMessage(client, TOPIC_DATA_UP, &pubmsg, &token);
                rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
                printf("Message with delivery token %d delivered\n", token);
                flag = 0;
            }
            sleep(5); //必须要加，过快的循环会导致未被发现的内存泄漏爆炸！！！！！！！！！！！！！
        }             //能力有限，能跑就行，此处代码没有百分百的把握勿迭代！！！！！
    }
}

//按照周期循环将共享内存的数据存入数据库
void *handler_sql()
{
    //最后完善
    printf("保存历史数据");
    return NULL;
}

////////////////////////////////////////
//mqtt相关封装，勿动！！！！！！！！****************************************************************************
volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;
    char get_buf[256] = {0}; //存放接收json的内容
    //printf("Message arrived\n");
    //printf("     topic: %s\n", topicName);
    //printf("   message: ");

    //逻辑笔记：回调函数中会获得接收的主题，主题内容这两个信息，首先判断主题类型：控制命令、模式修改、刷新上报、历史数据请求
    //写四个if else
    //明天需要研究框架自带的消息队列库，用于控制命令的判断逻辑中

    payloadptr = message->payload;
    for (i = 0; i < message->payloadlen; i++)
    {
        //putchar(*payloadptr++);
        get_buf[i] = *payloadptr;
        payloadptr++;
    }
    //putchar('\n');
    printf("topic: %s\n", topicName);
    printf("%s\n", get_buf);
    //判断主题
    
    if (strcmp(topicName, "/app/data/down") == 0)
    {
        char send_buf[2048] = {0}; //回复
        /* 数据上报、历史数据上报 */
        //判断是数据上报还是历史数据上报,json反序列化
        cJSON *root = NULL;
        root = cJSON_Parse(get_buf);
        int up_type = cJSON_GetObjectItem(root, "type")->valueint;
        if (up_type == 1)
        {
            //刷新上报
            snprintf(send_buf, 2048, "%s", get_all_data());
            //printf("%s\n", send_buf);

            pubmsg.payload = send_buf;
            pubmsg.payloadlen = (int)strlen(send_buf);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, TOPIC_DATA_UP, &pubmsg, &token);
            rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
        }
        else if (up_type == 4)
        {
            //上报历史数据
        }
    }
    else if (strcmp(topicName, "/app/control/down") == 0)
    {
        /* 控制命令处理，传感器等硬件的控制命令，分类后存入各自的消息队列 */
        //获取传入json里的key值，判断是stm32的命令还是modbus的命令
        cJSON *root = NULL;
        cJSON *item = NULL;
        root = cJSON_Parse(get_buf);
        item = cJSON_GetObjectItem(root, "data");
        int key = cJSON_GetObjectItem(item, "key")->valueint;
        printf("%d\n", key);
        if (key > 300)
        {
            printf("stm32\n");
            snprintf(send_buf.mdata, 256, "%s", get_buf);
            if (msg_queue_send("stm32", &send_buf, sizeof(send_buf), 0) < 0)
            {
                printf("msg_queue_send error\n");
                return -1;
            }
        }
        else if (key < 300 && key > 100)
        {
            printf("modbus\n");
            snprintf(send_buf.mdata, 256, "%s", get_buf);
            if (msg_queue_send("modbus", &send_buf, sizeof(send_buf), 0) < 0)
            {
                printf("msg_queue_send error\n");
                return -1;
            }
        }
        //控制命令存入各自的消息队列后应该给QT客户端返回一个包含控制成功的json，此处省略
    }
    else if (strcmp(topicName, "/app/mode/down") == 0)
    {
        /* 上报模式更改 */
        //获取传入json里的data，将data中的两个参数修改进json文件中
        cJSON *root = NULL;
        cJSON *item = NULL;
        root = cJSON_Parse(get_buf);
        item = cJSON_GetObjectItem(root, "data");
        int type = cJSON_GetObjectItem(item, "type")->valueint;
        int period = cJSON_GetObjectItem(item, "period")->valueint;
        //读取json文件
        int fd;
        char buf[2048] = {0};
        fd = open("/mnt/config/node.json", O_RDONLY);
        if (fd < 0)
        {
            perror("error");
            return -1;
        }
        int read_len = read(fd, buf, 2048);
        if (read_len < 0)
        {
            perror("error");
            return -1;
        }
        close(fd);
        //创建一个json对象
        cJSON *file_root = NULL;
        cJSON *file_item = NULL; //cjson对象
        file_root = cJSON_Parse(buf);
        file_item = cJSON_GetObjectItem(file_root, "report");
        //cJSON_GetObjectItem(file_item, "type")->valueint = type;
        //cJSON_GetObjectItem(file_item, "period")->valueint = period;
        //cJSON_AddItemToObject(file_item, "type", cJSON_CreateNumber(type));
        //cJSON_AddItemToObject(file_item, "period", cJSON_CreateNumber(period));
        cJSON_SetIntValue(cJSON_GetObjectItem(file_item, "type"),type);
        cJSON_SetIntValue(cJSON_GetObjectItem(file_item, "period"),period);

        fd = open("/mnt/config/node.json", O_RDWR | O_TRUNC);
        if (fd < 0)
        {
            perror("error");
            return -1;
        }
        printf("%s\n",cJSON_Print(file_root));
        char new_json[2048] = {0};
        snprintf(new_json,2048,"%s",cJSON_Print(file_root));
        int write_len = write(fd, new_json, sizeof(new_json));
        if(write_len < 1)
        {
            perror("error");
            return -1;
        }
        close(fd);
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
//mqtt相关封装，勿动！！！！！！！！****************************************************************************
/////////////////////////////////////////////

//指令接收线程，对指令的分类，存入对应的消息队列等操作皆在回调函数 msgarrvd 中
void *handler_order()
{
    //这个线程没有存在的必要了～
    return NULL;
}

int main(int argc, char const *argv[])
{
    //主进程开始，连接mqtt服务，建立数据库，解析点表{?},映射内存并初始化设备参数的数值
    //建立mqtt链接********************************************************
    //conn_opts = MQTTClient_connectOptions_initializer;
    //pubmsg = MQTTClient_message_initializer;
    init_reporter();
    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, "/app/+/down", QOS); //加号是通配符，这种写法可以订阅三个down主题，不可连续使用三个订阅函数，会刷新订阅主题数据

    //建立数据库***********************************************************
    //sqlite3 *db;
    //char *errmsg;
    //if (sqlite3_open("./data.db", &db))
    //{
    //    fprintf(stderr, "sqlite3_open err:%s\n", sqlite3_errmsg(db));
    //    return -1;
    //}

#if 0
    if(sqlite3_exec(db,"create table stu(id int primary key,name char,score float);",NULL,NULL,&errmsg))
	{
		fprintf(stderr,"create err:%s\n",errmsg);
		return -1;
	}
#endif

    //sqlite3_close(db);

    //映射内存并初始化设备参数****************************************************
    int ret = -1;
    ret = shm_init(&para_modbus, "modbus", 1024); //共享内存名，需要对接
    if (ret < 0)
    {
        return -1;
    }
    ret = shm_init(&para_stm32, "stm32", 1024); //共享内存名，需要对接
    if (ret < 0)
    {
        return -1;
    }
    modbus_addr = shm_getaddr(&para_modbus);
    stm32_addr = shm_getaddr(&para_stm32);
    if ((modbus_addr == NULL) || (stm32_addr == NULL))
    {
        return -1;
    }

    //启动 上报线程 数据存储线程 指令处理线程
    pthread_t tid;
    if (pthread_create(&tid, NULL, handler_up, NULL) != 0) //上报线程
    {
        printf("handler_up create err");
        return -1;
    }
    if (pthread_create(&tid, NULL, handler_sql, NULL) != 0) //数据存储线程
    {
        printf("handler_sql create err");
        return -1;
    }
    if (pthread_create(&tid, NULL, handler_order, NULL) != 0) //指令处理线程
    {
        printf("handler_order create err");
        return -1;
    }

    pid_t pid;
    int i;
    for (i = 0; i < 7; ++i)
    {
        pid = fork();
        if (pid == -1)
        {
            perror("fork失败!");
            exit(1);
        }

        //循环中，fork函数调用五次，子进程返回0,父进程返回子进程的pid，
        //为了避免子进程也fork，需要判断并break
        if (pid == 0)
        {
            break;
        }
    }

    //返回大于0的进程就是父进程
    if (pid > 0) //父进程
    {
        //printf("父进程: pid= %d , ppid=%d,子进程: %d i的变量是: %d\n", getpid(), getppid(), pid, i);
        while (1)
            ; //这里延迟父进程程序，等子进程先执行完。
    }
    else if (pid == 0) //子进程
    {
        //printf("i的变量是: %d 子进程: pid= %d , ppid=%d \n", i, getpid(), getppid());
        if (i == 0)
        {
            //MODBUS数据采集进程
            modbus_init();
        }
        else if (i == 1)
        {
            //MODBUS指令下发进程
        }
        else if (i == 2)
        {
            //STM32数据采集进程
        }
        else if (i == 3)
        {
            //STM32数据下发进程
        }
        else if (i == 4)
        {
            //数据上报进程
            //init_reporter();   //网关点表初始化
            //data_reporter();
            //data_cycle_reporter();
        }
        else if (i == 5)
        {
            //待补充进程进程
        }
        else if (i == 6)
        {
        }
    }

    pthread_join(tid, NULL);
    while (1)
        ;
    return 0; //养成好习惯，返回程序执行状态码
}
